/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tablet;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.io.NWritable;
import com.nhncorp.neptune.common.io.NWritableUtils;

/**
 * 특정 row의 특정 column에 저장되어 있는 모든 값을 가진다(ValueObject).<br>
 * @author 김형준
 *
 */
public class RowColumnValues implements NWritable, Comparable<RowColumnValues>, org.apache.hadoop.io.Writable {
	private Row.Key rowKey;
	private String columnName;
  private ColumnValue[] columnValues;
	
	public RowColumnValues() {
		rowKey = new Row.Key();
	}

	public RowColumnValues(String columnName, Row.Key rowKey, ColumnValue[] columnValues) {
		this.rowKey = rowKey;
		this.columnName = columnName;
		this.columnValues = columnValues;
	}
	
	public void readFields(DataInput in) throws IOException {
		rowKey.readFields(in);
		columnName = NWritableUtils.readString(in);
		
		int recordCount = in.readInt();
		
		columnValues = new ColumnValue[recordCount];
		for(int i = 0; i < recordCount; i++) {
      int flag = in.readInt();
      if(flag == 0) {
        columnValues[i] = null;
      } else {
  			columnValues[i] = new ColumnValue();
  			columnValues[i].readFields(in);
      }
		}
	}

	public void write(DataOutput out) throws IOException {
		rowKey.write(out);
		NWritableUtils.writeString(out, columnName);

		int recordCount = (columnValues == null ? 0 : columnValues.length);
		out.writeInt(recordCount);
		
		for(int i = 0; i < recordCount; i++) {
      if(columnValues[i] == null) {
        out.writeInt(0);
      } else {
        out.writeInt(1);
        columnValues[i].write(out);
      }
		}
	}

	public String getColumnName() {
		return columnName;
	}

	public Row.Key getRowKey() {
		return rowKey;
	}

	public ColumnValue[] getColumnRecords() {
		return columnValues;
	}

	public void setColumnRecords(ColumnValue[] columnValues) {
		this.columnValues = columnValues;
	}

	public void setColumnName(String columnName) {
		this.columnName = columnName;
	}

	public void setRowKey(Row.Key rowKey) {
		this.rowKey = rowKey;
	}

//  public void merge(List<RowColumnValues> mergeTargetColumnResults) {
//    //자기 자신도 merge 대상에 포함시킴
//    mergeTargetColumnResults.add(this);
//    
//    Map<Cell.Key, ValueCollection> columnMap = new TreeMap<Cell.Key, ValueCollection>();
//    
//    for(RowColumnValues eachColumnResult: mergeTargetColumnResults) {
//      for(ColumnValue eachColumnRecord: eachColumnResult.getColumnRecords()) {
//        Cell.Key columnKey = eachColumnRecord.getCell.Key();
//        ValueCollection valueCollection = columnMap.get(columnKey);
//        if(valueCollection == null) {
//          valueCollection = new ValueCollection();
//          columnMap.put(columnKey, valueCollection);
//        }
//        valueCollection.add(eachColumnRecord);
//      }
//    }
//    
//    List<ColumnValue> mergeResult = new ArrayList<ColumnValue>(10);
//    for(ValueCollection valueCollection: columnMap.values()) {
//      //여기서 삭제된 데이터 또는 버전에 대한 처리를 수행한다.
//      ColumnValue eachRecord = valueCollection.get();
//      if(eachRecord != null) {
//        mergeResult.add(eachRecord);
//      }
//    }
//    
//    columnValues = mergeResult.toArray(new ColumnValue[mergeResult.size()]);
//  }

	public int compareTo(RowColumnValues rowColumnValues) {
//	  String str1 = columnName + rowKey.toString();
//		String str2 = rowColumnValues.columnName + rowColumnValues.rowKey.toString();
//		return str1.compareTo(str2);

    String str1 = rowKey.toString() + columnName;
    String str2 = rowColumnValues.rowKey.toString() + rowColumnValues.columnName;
    return str1.compareTo(str2);
	}

  public boolean equals(Object obj) {
    if( !(obj instanceof RowColumnValues) ) {
      return false;
    }
    
    return compareTo((RowColumnValues)obj) == 0;
  } 
  
	public void print() {
		if(rowKey == null || rowKey.getBytes() == null) {
			System.out.println("empty result");
			return;
		}
		
		System.out.println("Row.Key:" + rowKey + ", ColumnName:" + columnName + ", count:" + columnValues.length);
		if(columnValues == null || columnValues.length == 0) {
			System.out.println("\tempty");
			return;
		}
		for(ColumnValue columnValue: columnValues) {
			System.out.println("\t" + columnValue);
		}
	}
  
  public String toString() {
    return "Row.Key:" + rowKey + ", ColumnName:" + columnName + ", count:" + columnValues.length;
  }
}
